package prictice.transaction1.daily.partition;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.coordination.IBatchBolt;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.transactional.TransactionAttempt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import prictice.base.DateFmt;

import java.util.HashMap;
import java.util.Map;

/**
 * @Package prictice.transaction1.daily
 * @Description:
 * @Author elwyn
 * @Date 2017/9/17 16:21
 * @Email elonyong@163.com
 */
public class MyDailyBatchBolt implements IBatchBolt<TransactionAttempt> {
    private static final long serialVersionUID = -8719524746245495176L;
    BatchOutputCollector batchOutputCollector;
    TransactionAttempt tx;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, BatchOutputCollector batchOutputCollector, TransactionAttempt transactionAttempt) {
        this.batchOutputCollector = batchOutputCollector;
    }

    Integer count = null;
    String today = null;
    Map<String, Integer> countMap = new HashMap<>();

    @Override
    public void execute(Tuple tuple) {
        String log = tuple.getString(1);
        tx = (TransactionAttempt) tuple.getValue(0);
        if (log != null) {
            String[] split = log.split("\t");
            if (split.length >= 3) {
                today = DateFmt.getCountDate(split[2], DateFmt.date_short);
                count = countMap.get(today);
                if (count == null) {
                    count = 0;
                }
                count++;
                countMap.put(today, count);
            }
        }

    }

    @Override
    public void finishBatch() {
        System.err.println("tx" + tx + "---- today " + today + " count" + count);
        batchOutputCollector.emit(new Values(tx, today, count));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("tx", "date", "count"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
